import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.*;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Hdfs操作类
 *
 * @author Ace Lee
 * @date 2019/8/27 14:18
 * @version 1.0
 **/
public class HdfsFileSystem {

    /**
     * HDFS集群路径
     */
    private static final String hdfsPath = "hdfs://10.10.10.10:8020";

    public static void copyFileToHDFSByName(Configuration conf,String localFileName, String remoteFileName) throws IOException {
        FileSystem fs = FileSystem.get(URI.create(hdfsPath), conf);
        fs.copyFromLocalFile(new Path(localFileName), new Path(remoteFileName));
        System.out.println("copy from local file:" + localFileName + " to HDFS file:" + remoteFileName + " done.");
        fs.close();
    }

    public static void copyFileToHDFSByFileObj(byte[] bytes, String prefix, String fileName) throws IOException {
        InputStream in = null;
        try {
            Configuration conf = new Configuration();
            FileSystem fileSystem = FileSystem.get(URI.create(hdfsPath), conf);
            FSDataOutputStream out = fileSystem.create(new Path(buildPath(hdfsPath, "/document", prefix, fileName)));
            in = new ByteArrayInputStream(bytes);
            IOUtils.copyBytes(in, out, 4096, false);
            out.hsync();
            out.close();
        } finally {
            IOUtils.closeStream(in);
        }
        return;
    }

    public static void copyFileToHDFSByFileObj(String filename,String hdfsUri) throws IOException {
        if (null == hdfsUri || hdfsUri.isEmpty()) {
            System.err.println("copyFileToHDFSByFile: hdfsUri are required");
            return;
        }
        String localPath = HdfsFileSystem.class.getResource("").getPath();
        String localFile = localPath+filename;

        InputStream in = new FileInputStream(localFile);
        try {
            Configuration conf = new Configuration();
            FileSystem fileSystem = FileSystem.get(URI.create(hdfsPath), conf);
            FSDataOutputStream out = fileSystem.create(new Path(hdfsPath+hdfsUri+filename));
            IOUtils.copyBytes(in, out, 4096, false);
            out.hsync();
            out.close();
        } finally {
            IOUtils.closeStream(in);
        }
        return;
    }

    public static void copyFileToHDFSByFileObj(InputStream in, String prefix, String fileName) throws IOException {
        try {
            Configuration conf = new Configuration();
            FileSystem fileSystem = FileSystem.get(URI.create(hdfsPath), conf);
            FSDataOutputStream out = fileSystem.create(new Path(buildPath(hdfsPath, "/document", prefix, fileName)));
            IOUtils.copyBytes(in, out, 4096, false);
            out.hsync();
            out.close();
        } finally {
            IOUtils.closeStream(in);
        }
        return;
    }
    public static void copyFileToHDFSByFileObj(File localPath) throws IOException {
        InputStream in = null;
        if (null == localPath) {
            System.out.println("copyFileToHDFSByFile: localpath are required");
            return;
        }
        try {
            Configuration conf = new Configuration();
            FileSystem fileSystem = FileSystem.get(URI.create(hdfsPath), conf);
            FSDataOutputStream out = fileSystem.create(new Path(hdfsPath));

            in = new BufferedInputStream(new FileInputStream(localPath));
            IOUtils.copyBytes(in, out, 4096, false);
            out.hsync();
            out.close();
            //in.close();
        } finally {
            IOUtils.closeStream(in);
        }
        return;
    }

    /*
     * Download hdfs file in URI to local file
     */
    public static void downloadFromHDFS(Configuration conf, String uri, String remoteFileName, String localFileName) throws IOException {
        Path path = new Path(remoteFileName);
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        fs.copyToLocalFile(path, new Path(localFileName));
        fs.close();
        System.out.println("downloading file from " + remoteFileName + " to " + localFileName + " succeed");
        return;
    }


    /*
     * Download hdfs file in URI to local file
     */
    public static void downloadFromHDFS(String uri, String HDFSFileName, OutputStream localFileOutPut) throws IOException {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), config);
        InputStream is = fs.open(new Path(uri + "/" + HDFSFileName));
        IOUtils.copyBytes(is, localFileOutPut, 4096, true);//close in and out stream via this API itself.
        System.out.println("downloading HDFS file " + HDFSFileName + " succeed");
        fs.close();
        return;
    }

    public static InputStream downloadFromHDFS(String uri, String HDFSFileName) throws IOException {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), config);
        InputStream is = fs.open(new Path(uri + HDFSFileName));
        if (is == null) {
            System.out.println("hdfs inputStream is null");
        }
        return is;
    }

    /*
     * check whether the HDFS file exists in given URI
     */
    public static boolean exists(String HDFSUri, String HDFSFileName) {
        Configuration conf = new Configuration();
        boolean fileExists = false;
        try {
            FileSystem fileSystem = FileSystem.get(URI.create(HDFSUri), conf);
            fileExists = fileSystem.exists(new Path(HDFSUri + "/" + HDFSFileName));
        } catch (IOException e) {
            System.out.println("hdfs:exist() exception occurs. exception:" + e.getMessage());
            return fileExists;
        }

        System.out.println("HDFS URI:" + HDFSUri + ", fileName:" + HDFSFileName + " exists ? " + fileExists);
        return fileExists;
    }

    /**
     * 查看目录下面的文件
     *
     * @param uri
     * @param folder
     * @throws IOException
     */
    public static void ls(String uri, String folder) throws IOException {
        Configuration conf = new Configuration();
        Path path = new Path(folder);
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        FileStatus[] list = fs.listStatus(path);
        System.out.println("ls: " + folder);
        System.out.println("==========================================================");
        for (FileStatus f : list) {
            System.out.printf("name: %s, folder: %s, size: %d\n", f.getPath(), f.isDirectory(), f.getLen());
        }
        System.out.println("==========================================================");
        fs.close();
    }

    /**
     * 删除文件或者文件夹
     * @param uri
     * @param filePath
     * @throws IOException
     */
    public static void delete(String uri,String filePath) throws IOException {
        Configuration conf = new Configuration();
        Path path = new Path(filePath);
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        fs.deleteOnExit(path);
        System.out.println("Delete: " + filePath);
        fs.close();
    }


    public static String getCurrentDatePath(){
        return new SimpleDateFormat("yyyy/MM/dd/").format(new Date());
    }

    public static String getCurrentDateTime(){
        return new SimpleDateFormat("yyyyMMddHHmmssSSS").format(new Date());
    }

    public static String buildPath(String... paths){
        StringBuffer buffer=new StringBuffer(paths.length>0?paths[0]:"");
        if(paths!=null&&paths.length>=2){
            for(int i=1;i<paths.length;i++){
                if(paths[i]==null||paths[i].length()==0){
                    continue;
                }
                if(paths[i-1].endsWith("/")){
                    if(paths[i].startsWith("/")){
                        buffer.append(paths[i].substring(1));
                    }else{
                        buffer.append(paths[i]);
                    }
                }else{
                    if(paths[i].startsWith("/")){
                        buffer.append(paths[i]);
                    }else{
                        buffer.append("/").append(paths[i]);
                    }
                }

            }
        }
        return buffer.toString();
    }


    public static String writeFile(String data) throws Exception{
        String localPath = HdfsFileSystem.class.getResource("").getPath();
        String localFilename = getCurrentDateTime()+".txt";
        localPath+=localFilename;
        File file = new File(localPath);

        //if file doesnt exists, then create it
        if(!file.exists()){
            file.createNewFile();
        }

        FileWriter fileWritter = new FileWriter(file.getName());
        BufferedWriter bufferWritter = new BufferedWriter(fileWritter);
        bufferWritter.write(data);
        bufferWritter.close();
        fileWritter.close();
        System.out.println(localPath+" [write] done");

        return localFilename;
    }

    public static void deleteFile(String filename) throws Exception{
        String localPath = HdfsFileSystem.class.getResource("").getPath();
        File file = new File(localPath+filename);
        if(file.isFile() && file.exists()){
            boolean delete = file.delete();
            System.out.println(file.getPath()+" [delete] "+delete);
        }
    }
}
